package com.poetic.order.function;

import com.poetic.order.domain.OrderEvent;
import com.poetic.order.domain.OrderResult;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class OrderPayMatch extends KeyedProcessFunction<String, OrderEvent, OrderResult> {
    private OutputTag<OrderResult> orderTimeoutOutputTag;

    public OrderPayMatch(OutputTag<OrderResult> orderTimeoutOutputTag) {
        this.orderTimeoutOutputTag = orderTimeoutOutputTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void processElement(OrderEvent value, Context ctx, Collector<OrderResult> out) throws Exception {

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderResult> out) throws Exception {
        super.onTimer(timestamp, ctx, out);
    }

    /**
     * lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("ispayed-state", classOf[Boolean]))
     *     // 保存定时器的时间戳为状态
     *     lazy val timerState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-state", classOf[Long]))
     *
     *     override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {
     *       // 先读取状态
     *       val isPayed = isPayedState.value()
     *       val timerTs = timerState.value()
     *
     *       // 根据事件的类型进行分类判断，做不同的处理逻辑
     *       if( value.eventType == "create" ){
     *         // 1. 如果是create事件，接下来判断pay是否来过
     *         if( isPayed ){
     *           // 1.1 如果已经pay过，匹配成功，输出主流，清空状态
     *           out.collect( OrderResult(value.orderId, "payed successfully") )
     *           ctx.timerService().deleteEventTimeTimer(timerTs)
     *           isPayedState.clear()
     *           timerState.clear()
     *         } else {
     *           // 1.2 如果没有pay过，注册定时器等待pay的到来
     *           val ts = value.eventTime * 1000L + 15 * 60 * 1000L
     *           ctx.timerService().registerEventTimeTimer(ts)
     *           timerState.update(ts)
     *         }
     *       } else if ( value.eventType == "pay" ){
     *         // 2. 如果是pay事件，那么判断是否create过，用timer表示
     *         if( timerTs > 0 ){
     *           // 2.1 如果有定时器，说明已经有create来过
     *           // 继续判断，是否超过了timeout时间
     *           if( timerTs > value.eventTime * 1000L ){
     *             // 2.1.1 如果定时器时间还没到，那么输出成功匹配
     *             out.collect( OrderResult(value.orderId, "payed successfully") )
     *           } else{
     *             // 2.1.2 如果当前pay的时间已经超时，那么输出到侧输出流
     *             ctx.output(orderTimeoutOutputTag, OrderResult(value.orderId, "payed but already timeout"))
     *           }
     *           // 输出结束，清空状态
     *           ctx.timerService().deleteEventTimeTimer(timerTs)
     *           isPayedState.clear()
     *           timerState.clear()
     *         } else {
     *           // 2.2 pay先到了，更新状态，注册定时器等待create
     *           isPayedState.update(true)
     *           ctx.timerService().registerEventTimeTimer( value.eventTime * 1000L )
     *           timerState.update(value.eventTime * 1000L)
     *         }
     *       }
     *     }
     *
     *     override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {
     *       // 根据状态的值，判断哪个数据没来
     *       if( isPayedState.value() ){
     *         // 如果为true，表示pay先到了，没等到create
     *         ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "already payed but not found create log"))
     *       } else{
     *         // 表示create到了，没等到pay
     *         ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "order timeout"))
     *       }
     *       isPayedState.clear()
     *       timerState.clear()
     *     }
     *   }
     */
}
